package com.example.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.kafka.config.KafkaStreamListener;
import com.example.kafka.constant.StreamConstant;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Predicate;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.Windowed;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 20:20
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Component
public class StreamHandlerListener implements KafkaStreamListener<KStream<String,String>> {

    public static Pattern pattern = Pattern.compile("[0-9]+");


    @Override
    public String inPutTopic() {
        return StreamConstant.IN_PUT_TOPIC;
    }

    @Override
    public String outPutTopic() {
        return StreamConstant.OUT_PUT_TOPIC;
    }

    @Override
    public KStream<String, String> handle(KStream<String, String> stream) {
        return  stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            /**
             * 把消息中的词组，转换为一个一个的单词放到集合中
             * @param value
             * @return
             */
            @Override
            public Iterable<String> apply(String value) {
                System.out.println("消息的value:"+value);//hello kafka stareams
                String[] strings = value.split("-");
                return Arrays.asList(strings);
            }
        }).filter(new Predicate<String, String>() {
                @Override
                public boolean test(String key,String value) {
                    return isNumber(value);
                }
            }
         ).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 把消息的key,重新赋值，
             * @param key
             * @param value
             * @return
             */
            @Override
            public KeyValue<String, String> apply(String key,String value) {
                return new KeyValue<>(value,value);
            }
        }).groupByKey()
                //时间聚合窗口
                .windowedBy(TimeWindows.of(5000))
                //消息的value就是聚合单词后的统计数值，long类型
                .count(Materialized.as("count-word-num-0001"))
                //转换为Kstream
                .toStream()
                //把处理后的key和value转换String
                .map((key,value)->{
                    return new KeyValue<>(key.key().toString(),value.toString());
                });
    }

    private static boolean isNumber(String str) {
        Matcher isNum = pattern.matcher(str);
        return isNum.matches();
    }
}
